package middleware.p2p;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.BindException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.URL;
import java.util.LinkedList;
import java.util.List;

import middleware.p2p.log.VectorClock;
import middleware.p2p.replication.ReplicaManager;
import middleware.p2p.replication.ReplicaManagerService;
import middleware.p2p.transaction.TransactionCoordinator;
import middleware.p2p.transaction.TransactionCoordinatorService;
import middleware.p2p.transaction.TransactionResponse;

import org.apache.log4j.Logger;

public class ReplicaFrontend {
	private TransactionCoordinator coordinator;
	private URL coordinatorWsdl;
	private VectorClock vclock=new VectorClock();
	private String thisAddr;
	public ReplicaFrontend(String thisAddr,URL coordWsdl) {
		this.thisAddr=thisAddr;		
		TransactionCoordinatorService service;
		coordinatorWsdl=coordWsdl;
		if (coordinatorWsdl!=null) {
			service = new TransactionCoordinatorService(coordinatorWsdl);
		} else {
			service= new TransactionCoordinatorService();
			Logger.getLogger(this.getClass()).debug("Missing coordinator WSDL URL!");
		}
		this.coordinator=service.getTransactionCoordinatorPort();
		for (WorkerDescriptor t:CORE.getWorkerDiscovery().getWorkers()) {
			this.vclock.setClock(CORE.generateComponentId(t.getAddr()), 0);
		}
		for (URL wsdl:CORE.getWorkerDiscovery().getAllRM()) {
			ReplicaManagerService ser=new ReplicaManagerService(wsdl);
			ReplicaManager rm=ser.getReplicaManagerPort();
			rm.resetClock(CORE.getComponentId());
		}
	}
	public boolean setTransactionCoordinatorWsdl(URL coordWsdl) {
		boolean retValue=true;
		if (!coordWsdl.equals(coordinatorWsdl)) {
			TransactionCoordinatorService service;
			if (coordWsdl!=null) {
				service = new TransactionCoordinatorService(coordWsdl);
			} else {
				service= new TransactionCoordinatorService();
				retValue=false;
			}
			this.coordinator=service.getTransactionCoordinatorPort();
		}
		return retValue;
	}
	public boolean storeReplica(String content) {		
		ReplicaStorerThread th=new ReplicaStorerThread(this.vclock, this.coordinator, this.coordinatorWsdl, content);
		th.start();
		return true;
	}
	public String getReplica(){
		String transId=this.coordinator.openTransaction();
		URL rmUrls[] = CORE.getWorkerDiscovery().getAllRM();
		LinkedList<String> responseList=new LinkedList<String>();
		ReplicaResponseHandler rrh=new ReplicaResponseHandler(rmUrls.length, Integer.parseInt(this.thisAddr.split(":")[1]), responseList);
		rrh.start();		
		String thisIp=this.thisAddr.split(":")[0];
		for (int k=0;k<rmUrls.length;k++) {
			ReplicaManagerService service=new ReplicaManagerService(rmUrls[k]);
			ReplicaManager rm=service.getReplicaManagerPort();
			VectorClock localClock=this.vclock.addEventToClock(CORE.getComponentId());
			rm.getReplica(CORE.getComponentId(), transId,this.coordinatorWsdl.toString(),localClock,thisIp+":"+rrh.getPort());
		} 		
		try {
			rrh.join();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		TransactionResponse resp=this.coordinator.closeTransaction(transId);
		String replica=null;
		if (resp==TransactionResponse.COMMIT) {
			if (responseList.size()>0) { 
				replica=responseList.getFirst();
			}
		}		
		Logger.getLogger(this.getClass()).debug("Transaction Response : "+resp+" , Operation Results  : "+replica);
		return replica;
	}
	private class ReplicaStorerThread extends Thread {
		private VectorClock vclock;
		private TransactionCoordinator coordinator;
		private String content;
		private URL coordinatorWsdl;
		public ReplicaStorerThread(VectorClock vc,TransactionCoordinator coord,URL wsdl,String cont) {
			this.vclock=vc;
			this.coordinator=coord;
			this.content=cont;
			this.coordinatorWsdl=wsdl;
			
		}
		public void run() {
			String transId=this.coordinator.openTransaction();
			URL rmUrls[]=CORE.getWorkerDiscovery().getAllRM();
			boolean result=true;
			LinkedList<String> responseList=new LinkedList<String>();
			ReplicaResponseHandler rrh=new ReplicaResponseHandler(rmUrls.length, Integer.parseInt(thisAddr.split(":")[1])+1, responseList);
			rrh.start();
			String thisIp=thisAddr.split(":")[0];
			for (int k=0;k<rmUrls.length;k++) {
				ReplicaManagerService service=new ReplicaManagerService(rmUrls[k]);
				ReplicaManager rm=service.getReplicaManagerPort();
				VectorClock localClock=this.vclock.addEventToClock(CORE.getComponentId());
				//System.out.println("===>storing ..."+transId+" on "+this.coordinatorWsdl.toString()+"\n"+localClock.toString());				
				result=result&&rm.storeReplica(CORE.getComponentId(), content, transId, this.coordinatorWsdl.toString(),localClock,thisIp+":"+rrh.getPort());
			}
			try {
				rrh.join();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			//System.out.println("END JOIN >.<");
			TransactionResponse resp=this.coordinator.closeTransaction(transId);
			Logger.getLogger(this.getClass()).debug("Transaction Response : "+resp+" , Operation Results (And Combination) : "+result);
		}
	}
	private class ReplicaResponseHandler extends Thread {
		private int numSock;
		private int port;
		final private List<String> responseList;
		public ReplicaResponseHandler(int numSock, int port,
				List<String> responseList) {
			this.numSock = numSock;
			this.port = port;
			this.responseList = responseList;
		}
		public int getPort() {
			return port;
		}
		public void run() {
			ServerSocket serv=null;
			while (serv==null) {
				try {
					serv = new ServerSocket(port);
				} catch (BindException be) {
					port++;
				}catch (IOException e2) {
				// 	TODO Auto-generated catch block
					e2.printStackTrace();
				}
			}
			for (int i=0;i<this.numSock;i++) {
				try {
					final Socket incoming = serv.accept();				
					Thread th=new Thread(new Runnable() {					
						public void run() {
							try {
								BufferedReader inSock=new BufferedReader(new InputStreamReader(incoming.getInputStream()));
								responseList.add(inSock.readLine());
								inSock.close();
								incoming.close();
							} catch (IOException e) {
								// TODO Auto-generated catch block
								e.printStackTrace();
							}
						}
					});
					th.start();
				} catch (IOException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
			}
		}		
	}
}
